package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.CourseReviewBean;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.DimAsyncFunction;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.TimeUnit;


/*

//TODO doris建表
        drop table if exists dws_course_review_count_window;
        create table if not exists dws_course_review_count_window
        (
        `stt`                       DATETIME comment '窗口起始时间',
        `edt`                       DATETIME comment '窗口结束时间',
        `cur_date`                  DATE comment '当天日期',
        `course_id`                 STRING comment '课程ID',
        `course_name`               STRING comment '课程名称',
        `unique_user_count`         BIGINT replace comment '评论用户数',
        `review_count` 		        BIGINT replace comment '评论次数',
        `high_review_count` 		BIGINT replace comment '好评次数',
        `grade_count` 		        BIGINT replace comment '总评分'
        ) engine = olap aggregate key (`stt`, `edt`, `cur_date`,`course_id`)
        comment "课程主题课程评价各窗口汇总表"
        partition by range(`cur_date`)()
        distributed by hash(`stt`) buckets 10 properties (
        "replication_num" = "3",
        "dynamic_partition.enable" = "true",
        "dynamic_partition.time_unit" = "DAY",
        "dynamic_partition.start" = "-1",
        "dynamic_partition.end" = "1",
        "dynamic_partition.prefix" = "par",
        "dynamic_partition.buckets" = "10",
        "dynamic_partition.hot_partition_num" = "1"
        );


*/



public class DwsCourseReviewCountWindow_lyy extends BaseAppOneStream {
    public static void main(String[] args) {
        new DwsCourseReviewCountWindow_lyy().init(
                6001,
                1,
                "DwsCourseReviewCountWindow",
                Constant.TOPIC_DWD_COURSE_REVIEW_INFO

        );

    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {

        //stream.print();

        //1、转换成POJO类
         SingleOutputStreamOperator<CourseReviewBean> ReviewBeanStream= TransReviewPojo(stream);

         //2、 判断，userId分组，把每组第一条数据标记为1，
        SingleOutputStreamOperator<CourseReviewBean> UniqueStream = UniqueReview(ReviewBeanStream);

        //3、开窗聚合
       SingleOutputStreamOperator<CourseReviewBean> ReviewStream = windowAndAgg(UniqueStream);

        //4.补充维度
        SingleOutputStreamOperator<CourseReviewBean> afterAddDimStream = addDim(ReviewStream);
      //  afterAddDimStream.print("oooooooooo");

       //5、写出到doris
      //  writeToDoris(afterAddDimStream);

    }

    private void writeToDoris(SingleOutputStreamOperator<CourseReviewBean> reviewStream) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_course_review_count_window";

        reviewStream
                .map(pojo -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(pojo,config);
                })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));
    }

    //添加维度
    private SingleOutputStreamOperator<CourseReviewBean> addDim(SingleOutputStreamOperator<CourseReviewBean> reviewStream) {

        return AsyncDataStream.unorderedWait(
                reviewStream,
                new DimAsyncFunction<CourseReviewBean>() {
                    @Override
                    public String getTable() {
                        return "DIM_COURSE_INFO";
                    }

                    @Override
                    public String getId(CourseReviewBean input) {
                        return input.getCourseId();  //pojo中得CourseId
                    }

                    @Override
                    public void addDim(JSONObject dim, CourseReviewBean input) {
                        input.setCourseName(dim.getString("COURSE_NAME"));

                    }
                },
                60,
                TimeUnit.SECONDS
        );
}

    private SingleOutputStreamOperator<CourseReviewBean> windowAndAgg(
            SingleOutputStreamOperator<CourseReviewBean> uniqueStream) {
        return uniqueStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<CourseReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner( //为数据流的每个元素设置时间戳
                        (bean,ts) -> bean.getTs()*1000
                )
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(
                        new ReduceFunction<CourseReviewBean>() {
                            @Override
                            public CourseReviewBean reduce(CourseReviewBean bean1,
                                                           CourseReviewBean bean2) throws Exception {
                                bean1.setUniqueUserCount(bean1.getUniqueUserCount() + bean2.getUniqueUserCount());
                                bean1.setHighReviewCount(bean1.getHighReviewCount() + bean2.getHighReviewCount());
                                bean1.setGradeCount(bean1.getGradeCount() + bean2.getGradeCount());
                                bean1.setReviewCount(bean1.getReviewCount() + bean2.getReviewCount());

                                return bean1;
                            }
                        },
                        new ProcessAllWindowFunction<CourseReviewBean, CourseReviewBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx,
                                                Iterable<CourseReviewBean> elements,
                                                Collector<CourseReviewBean> out) throws Exception {
                                CourseReviewBean bean = elements.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDate(ctx.window().getStart()));

                                out.collect(bean);
                            }
                        }
                );


    }

    private SingleOutputStreamOperator<CourseReviewBean> UniqueReview(
            SingleOutputStreamOperator<CourseReviewBean> reviewBeanStream) {
        return reviewBeanStream
                .keyBy(CourseReviewBean::getUserId)
                .process(new KeyedProcessFunction<String, CourseReviewBean, CourseReviewBean>() {
                    private ValueState<String> uniqueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        uniqueState= getRuntimeContext().getState(
                                new ValueStateDescriptor<String>("UniqueState",String.class));
                    }

                    @Override
                    public void processElement(CourseReviewBean bean,
                                               Context ctx,
                                               Collector<CourseReviewBean> out) throws Exception {
                        if(uniqueState.value() == null){
                            bean.setUniqueUserCount(1L); //用户评论数
                            uniqueState.update("true");
                        }
                        out.collect(bean);

                    }
                });


    }

    private SingleOutputStreamOperator<CourseReviewBean> TransReviewPojo(DataStreamSource<String> stream) {
        return
                stream.map(json -> {
                    JSONObject obj = JSON.parseObject(json);

                    //user 用来分组判断独立用户数
                    String user_id = obj.getString("user_id");

                    long review_stars = Long.parseLong(obj.getString("review_stars"));

                    long ts = Long.parseLong(obj.getString("ts"));

                    //判断是否好评
                    long HighReviewCount = 0L;
                    if (obj.getString("review_stars") == "5") {
                        HighReviewCount = 1L;

                    }
                    return new CourseReviewBean(
                            "",
                            "",
                            "",
                            "",
                            "",
                            "",
                            0L,
                            1L,
                            HighReviewCount,
                            review_stars,
                            ts
                    );
                });

    }
}

